package qsort;


import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.PriorityQueue;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;




public class MultiThreadQSort {
	int NUMBER_OF_KEYS = 0 ;
	int NUMBER_OF_THREADS = 0;
	int BLOCK_SIZE = 100000;
	long START_TIME = 0 ;
	String OUT_FILE = ""; 
	ArrayList<Record> SortList = new ArrayList<Record>();
	
	// Constructors 
	public MultiThreadQSort(){
		NUMBER_OF_KEYS = 1; 
		NUMBER_OF_THREADS = 1;
		OUT_FILE = "Result.txt"; // Default out file 
	}
	
	public MultiThreadQSort(int NumberOfKeys, int NumberOfThreads, String outFile){
		NUMBER_OF_KEYS = NumberOfKeys; 
		NUMBER_OF_THREADS = NumberOfThreads;
		OUT_FILE = outFile;
	}

	// Load input data and process to list of records  
	public void loadData(String inputFile) throws IOException{
		FileInputStream fis = new FileInputStream(inputFile); 
		BufferedReader br = new BufferedReader(new InputStreamReader(fis,"UTF8"));
		System.out.println(" Reading input file " + inputFile);
		String line= ""; 
		while ((line=br.readLine()) !=null){
			line = line.trim();
			String[] tokens = line.split("\\s+",NUMBER_OF_KEYS+1);
			// Build keys 
			ArrayList<String> keys = new ArrayList<String>();
			for (int i=0; i<NUMBER_OF_KEYS; i++){
				keys.add(tokens[i]);
			}
			// Build Record
			Record aLine = new Record(line, keys);
			// Add to list of record 
			SortList.add(aLine);
		}
		br.close(); 
		fis.close();
	}
	public ArrayList<Record> MultiThreadsQuickSort(int start, int end) throws InterruptedException{
		// Divide the data from start to end to X THREADS, proceed the thread and return shorted array
		ExecutorService es = Executors.newCachedThreadPool();
		int total = end - start + 1; 
		int eachThreadNeedToRun = total / NUMBER_OF_THREADS; 
		// Save the position of each threads 
		ArrayList<Entry<Integer,Integer>> savePosition = new ArrayList<Entry<Integer,Integer>>();
		
		for (int i =1; i<=NUMBER_OF_THREADS; i++){
			
			int fromStart = (i-1)* eachThreadNeedToRun ;
			int fromEnd = fromStart + eachThreadNeedToRun-1;
			
			if (i == NUMBER_OF_THREADS){ // last one => need to adjust the end 
				fromEnd = end; 
			}
			System.out.println(" --------------------  ");
			System.out.println(" Thread " + i + " from " + fromStart +" --> " + fromEnd );
			Entry<Integer,Integer> aBlock = new AbstractMap.SimpleEntry<Integer,Integer>(fromStart, fromEnd);
			savePosition.add(aBlock);
			QuickSort t = new QuickSort(SortList,fromStart, fromEnd);
			es.execute(t); 
		}
		// Wait till all the process is finished 
		es.shutdown();
		es.awaitTermination(Long.MAX_VALUE, TimeUnit.HOURS);

		
		// Join to form the final one using priority queue, complexity = O(n log m) : where m is number of threads			
		
		ArrayList<Record> result = new ArrayList<Record>();
		PriorityQueue<RecordAndListID> pQueue = new PriorityQueue<RecordAndListID>();
		// First enqueue all the first position 
		// Preserve nextNumber
		int[] nextNumber = new int[NUMBER_OF_THREADS];
		
		for (int i=0; i<savePosition.size(); i++){
			
			int startPosition = savePosition.get(i).getKey();
		
			// Enqueue (First record,thread ID) 
			pQueue.add(new RecordAndListID(SortList.get(startPosition), i));
			
			// Tracking the next number 
			nextNumber[i] = startPosition+1;
		}
		while (!pQueue.isEmpty()){
			// Get head 
			RecordAndListID head = pQueue.poll();
			
			// Output to result 
			result.add(head.getRecord());

			// Enqueue new items 
			int listID = head.getListID();
			int lastItemID = savePosition.get(listID).getValue();
			if (nextNumber[listID]<=lastItemID){
				pQueue.add(new RecordAndListID(SortList.get(nextNumber[listID]), listID));
				nextNumber[listID]++;
			}
		}

		return result; 
	}
		
	
	public ArrayList<Record> run() throws InterruptedException, IOException{
		return MultiThreadsQuickSort(0,SortList.size() -1);
	}
	
	public void outputResultFile(ArrayList<Record> result, String outFile) throws IOException{
		FileOutputStream fos = new FileOutputStream(outFile); 
		BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fos,"UTF8"));
		for (int i=0; i<result.size(); i++){
			bw.write(result.get(i).getData()+"\n");
		}
		bw.close(); 
		fos.close(); 				
	}
	public static void main(String[] args) throws IOException, InterruptedException{
 
		
		if (args.length != 4){
			System.out.println(" Usage : Source file + Key Number + Number of Threads + Output File");
			return; 
		}
		// Arguments 
		String inputFile = args[0];
		int NumberOfKeys = Integer.parseInt(args[1]);
		int NumberOfThreads = Integer.parseInt(args[2]);  
		String outputFile = args[3];
		
//		String inputFile = "data/test.txt";
//		int NumberOfKeys = 2;
//		int NumberOfThreads = 1;  
//		String outputFile = "resultQSort.tsv";
		
		//
		MultiThreadQSort temp = new MultiThreadQSort(NumberOfKeys, NumberOfThreads,outputFile);
		long TOTAL_START = System.currentTimeMillis();
		long startTime = System.currentTimeMillis();
		temp.loadData(inputFile); 
		long endTime = System.currentTimeMillis();
		System.out.println("\n TIME for loading data = " + (endTime - startTime) + " milliseconds");

		startTime = System.currentTimeMillis();
		ArrayList<Record> result = temp.run();
		endTime = System.currentTimeMillis();
		System.out.println("\n TIME for running quick sort = " + (endTime - startTime) + " milliseconds");
		
		startTime = System.currentTimeMillis();
		temp.outputResultFile(result, outputFile); 
		endTime = System.currentTimeMillis();
		System.out.println("\n TIME for output to file = " + (endTime - startTime) + " milliseconds");		
		
		endTime = System.currentTimeMillis();
		System.out.println("\n TOTAL TIME = " + (endTime - TOTAL_START) + " milliseconds");
		
	}
}
